package dse.messages.multiversion;

import java.util.*;

import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.Version;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;

public class GetDVSSVRTransaction extends Transaction implements SVRHandler {
	private DataObject dobj;
	private LinkedList<Node> residentNodes;
	private TreeSet<Version> versions;
	private Version curVersion;
	private HashMap<Node, Integer> verifyStatus;
	private int requestShareCount;
	private int validShareCount;
	private int maxFailure;
	private int round;
	private int successCount;
	private int failureCount;
	
	public GetDVSSVRTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		this.dobj = desiredObject;
		this.requestShareCount = dobj.shareCount();
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.validShareCount = this.requestShareCount-this.maxFailure;
		this.round = 1;
		this.successCount = 0;
		this.failureCount = 0;
	}
	
	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		if(requester.isDown()){
			System.err.println("SVR initiator is down!!!");
			return;
		}
		
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, dobj, experiment.modules().assigner());		
		versions = requester.getDataShare(dobj).getVersions();
		curVersion = versions.last();
		recorder.svrInitiated(super.id, requester, dobj, startTime);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "start", startTime);
		
		for(Node node : residentNodes)
			sendRequestVersionNumber(recorder, node, startTime);
	}
	
	private void sendRequestVersionNumber(ExperimentRecorder recorder, Node node, double time){
		Message m = super.createRequestVersionNumberMessage(requester, node, dobj, curVersion, round,
				MessagePurpose.SVR, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}

	@Override
	public void readVersion(ExperimentRecorder recorder, Node from,
			Version version, double time, int r) {
		// TODO Auto-generated method stub
		if(from.id() != requester.id()){
			if(version == null)
				return;
			if(verifyStatus == null)
				verifyStatus = new HashMap<Node, Integer>();
			if(verifyStatus.get(from) == null)
				verifyStatus.put(from, 1);
			else
				verifyStatus.put(from, verifyStatus.get(from) + 1);
			if(verifyStatus.get(from).intValue() == validShareCount){
				for(Node node : residentNodes)
					sendBoundaryVersion(recorder, from, node, time);
			}
			return;
		}
		
		if(finished || r<this.round)
			return;
		if(version == null){
			failureCount++;
			if(failureCount > maxFailure){
				curVersion = versions.lower(curVersion);
				if(curVersion != null){
					round++;
					successCount = 0;
					failureCount = 0;
					for(Node node : residentNodes){
						sendRequestVersionNumber(recorder, node, time);
					}
				}
				else
				{
					finished = true;
					if(ExperimentRecorder.debug)
						recorder.debug(
							"GetDVSSVRTransaction trans#" + super.id + 
							" is finshed!!!"
							);
					recorder.svrFinished(super.id, requester, dobj, time);
				}
			}
		}
		else{
			successCount++;
			if(successCount == validShareCount){
				successCount = 0;
				failureCount = 0;
				for(Node node : residentNodes)
					sendBoundaryVersion(recorder, requester, node, time);
			}
		}
	}
	
	private void sendBoundaryVersion(ExperimentRecorder recorder, Node from, Node to, double time){
		Message m = super.createBoundaryVersionMessage(from, to, dobj, curVersion, MessagePurpose.SVR, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	public void finishSVR(ExperimentRecorder recorder, double time, Node from){
		if(finished)
			return;
		successCount++;
		if(successCount >= validShareCount){
			finished = true;
			if(ExperimentRecorder.debug)
				recorder.debug(
					"GetDVSSVRTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.svrFinished(super.id, requester, dobj, time);
		}
	}
}
